package dse.messages;

import java.util.*;

import dse.*;
import dse.messages.locatedataobj.*;
import dse.messages.multiversion.*;
import dse.messages.singleversion.*;

/**
 * This is an abstract class to be used by other classes (the MessageHandler
 * classes and the Transaction classes) as a base class that provides a lot
 * of the common methods they use to send messages.
 * 
 * It is mostly a utility class, thats why it is so large.
 * 
 * @author Sam
 */
public abstract class MessageSender {
	
	protected Experiment experiment;
	//protected double ratio;
	
	
	protected MessageSender (Experiment experiment) {
		this.experiment = experiment;
		//ratio = experiment.configuration().getWritingReadingRatio();
	}
	
	//public double getRatio() { return ratio;}
	
	
	/**
	 * Get the communication costs between two nodes.  It returns the time to
	 * send the entire message and the distance between the nodes as a pair.
	 * 
	 * @param from
	 * @param to
	 * @param messageSize
	 * @return				The time to send the entire message and the distance
	 * 						between the nodes as a pair.
	 */
	protected Pair<Double,Double> getCommunicationCosts (
			Node from, 
			Node to, 
			long messageSize
			)
	{	
		double timeToSendEntireMessage = 
			experiment.modules().costCalculator().cost(
					experiment.network(),
					from, to, messageSize
					);
		// This should never fail
		assert timeToSendEntireMessage > 0; 
		double distanceBetweenNodes = 
			experiment.modules().costCalculator().distance(
					experiment.network(), from, to
					);
		// This should never fail too (we pretend like nodes at the same 
		// coordiantes have some minor distance).
		assert distanceBetweenNodes > 0;  
		return new Pair<Double, Double>(
				timeToSendEntireMessage,distanceBetweenNodes
				);
	}

	
	protected Message createRequestShareMessage(Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.RequestShare, from,
				to, MessagePurpose.DataObjectRequest,
				new RequestShareMessageHandler (
						experiment,
						dobj,
						shareIndex,
						MessagePurpose.DataObjectRequest,
						shareDownloader
						)
				);
	}
	
	
	/*protected Message createRequestShareLocalMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.RequestShareLocal, from,
				to, MessagePurpose.DataObjectRequest,
				new RequestShareLocalMessageHandler (
						experiment,
						dobj,
						shareIndex,
						MessagePurpose.DataObjectRequest,
						shareDownloader
						)
				);
	}*/
	
	protected Message createShareDataTransferMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			MessagePurpose purpose,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.ShareDataTransfer, from,
				to, dobj, purpose,
				new ShareDataTransferMessageHandler (
						experiment,
						dobj,
						shareIndex,
						shareDownloader
						)
				);
	}
	
	/*protected Message createHasShareRequestMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			MessagePurpose purpose,
			HasShareRequester hasShareRequester
			)
	{
		return createMessage(
				MessageType.HasShare, from,
				to, purpose, 
				new HasShareRequestMessageHandler (
						experiment,
						dobj,
						shareIndex,
						purpose,
						hasShareRequester
						)
				);
	}
	
	protected Message createHasShareResponseMessage (
			Node from, 
			Node to,
			boolean affermative,
			MessagePurpose purpose,
			HasShareRequester hasShareRequester
			)
	{
		MessageType t = 
			(affermative)?(MessageType.Affermative):(MessageType.Negative);
		
		return createMessage(
				t, from, to, purpose,
				new HasShareResponseMessageHandler (
						experiment,
						affermative,
						hasShareRequester
						)
				);
	}*/
	
	/*protected Message createHasDataObjectRequestMessage (
			Node from, 
			Node to,
			DataObject dobj,
			MessagePurpose purpose,
			HasDataObjectRequester hasShareRequester
			)
	{
		return createMessage(
				MessageType.HasDataObject, from,
				to, purpose, 
				new HasDataObjectRequestMessageHandler (
						experiment,
						dobj,
						purpose,
						hasShareRequester
						)
				);
	}
	
	protected Message createHasDataObjectResponseMessage (
			Node from, 
			Node to,
			boolean affermative,
			MessagePurpose purpose,
			HasDataObjectRequester hasShareRequester
			)
	{
		MessageType t = 
			(affermative)?(MessageType.Affermative):(MessageType.Negative);
		
		return createMessage(
				t, from, to, purpose,
				new HasDataObjectResponseMessageHandler (
						experiment,
						affermative,
						hasShareRequester
						)
				);
	}*/

	protected Message createRedirectShareDownloadMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			Node redirectTo,
			MessagePurpose purpose,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.Redirect, from, to, purpose, 
				new RedirectShareDownloadMessageHandler (
						experiment, dobj, shareIndex, redirectTo, 
						purpose, shareDownloader
						)
				);
	}
	
	/*protected Message createRedirectDataObjectDownloadMessage (
			Node from, 
			Node to,
			DataObject dobj,
			Node redirectTo,
			MessagePurpose purpose,
			DataObjectDownloader downloader
			)
	{
		return createMessage(
				MessageType.Redirect, from, to, purpose, 
				new RedirectDataObjectDownloadMessageHandler (
						experiment, dobj, redirectTo, purpose, downloader
						)
				);
	}*/

	protected Message createRequestShareForeignMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			MessagePurpose purpose,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.RequestShareForeign, from, to, purpose,
				new RequestShareForeignMessageHandler (
						experiment, dobj, shareIndex, purpose, shareDownloader
						)
				);
	}
	
	protected Message createNegativeRequestShareForeignResponseMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			MessagePurpose purpose,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.Negative, from, to, purpose,
				new NegativeRequestShareForeignResponseMessageHandler (
						experiment, dobj, shareIndex, shareDownloader
						)
				);
	}
	
	protected Message createLocateClustersWithDataObjectMessage (
			Node from, 
			Node to,
			DataObject dobj,
			MessagePurpose purpose,
			LocateClustersWithDataObjectRequester requester
			)
	{
		return createMessage(
				MessageType.LocateClustersWithDataObject, from, to, purpose,
				new LocateClustersWithDataObjectMessageHandler (
						experiment, dobj, purpose, requester
						)
				);
	}
	
	protected Message createClustersThatMightHaveDataObjectMessage (
			Node from, 
			Node to,
			LinkedList<Cluster> clusters,
			MessagePurpose purpose,
			LocateClustersWithDataObjectRequester requester
			)
	{
		return createMessage(
				MessageType.ClustersThatMightHaveDataObject, 
				from, 
				to, 
				clusters.size(),
				1, // for cluster id
				purpose,				
				new ClustersThatMightHaveDataObjectMessageHandler (
						experiment, clusters, requester
						)
				);
	}
	
	/*protected Message createCouldNotFindShareMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			MessagePurpose purpose,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.Negative, from,
				to, purpose,
				new CouldNotFindShareMessageHandler (
						experiment, dobj, shareIndex, shareDownloader
						)
				);
	}*/
	
	/*protected Message createCouldNotFindDataObjectMessage (
			Node from, 
			Node to,
			DataObject dobj,
			MessagePurpose purpose,
			DataObjectDownloader downloader
			)
	{
		return createMessage(
				MessageType.Negative, from, to, purpose,
				new CouldNotFindDataObjectMessageHandler (
						experiment, downloader
						)
				);
	}
	
	protected Message createBloomFilterRequestMessage (
			Node from, 
			Node to,
			BloomFilterRequester requester
			)
	{
		return createMessage(
				MessageType.BloomFilterRequest, from,
				to, MessagePurpose.BloomFilterRefresh,
				new BloomFilterRequestMessageHandler (
						experiment, requester
						)
				);
	}
	
	protected Message createBloomFilterDataTransferMessage (
			Node from, 
			Node to,
			BloomFilterRequester requester
			)
	{
		return createMessage(
				MessageType.BloomFilterDataTransfer, 
				from,
				to, 
				experiment.network().lookupCluster(to.clusterId()).
					bloomFilter(),
				MessagePurpose.BloomFilterRefresh,				
				new BloomFilterDataTransferMessageHandler (
						experiment, requester
						)
				);
	}
	
	protected Message createUpdateOnRecentLocalRequestsMessage (
			Node from, 
			Node to,
			SuperNodeLocalRequestUpdater updater
			)
	{
		return createMessage(
				MessageType.UpdateOnRecentLocalRequests, 
				from,
				to, 
				from.recentLocalRequestCount(),
				2, // for data object id + count
				MessagePurpose.Caching,				
				new UpdateRecentLocalRequestsMessageHandler (
						experiment, updater
						)
				);
	}
	
	protected Message createGetDataObjectInfoMessage (
			Node from, 
			Node to,
			DataObject dataObject,
			GetDataObjectInfoRequester requester
			)
	{
		return createMessage(
				MessageType.GetDataObjectInfo, 
				from,
				to, 
				MessagePurpose.Caching,				
				new GetDataObjectInfoMessageHandler (
						experiment, requester, dataObject
						)
				);
	}
	
	protected Message createDataObjectInfoResponseMessage (
			Node from, 
			Node to,
			boolean success,
			GetDataObjectInfoRequester requester
			)
	{
		return createMessage(
				((success)?
						(MessageType.DataObjectInfoResponse):
						(MessageType.Negative)), 
				from,
				to, 
				MessagePurpose.Caching,				
				new GetDataObjectInfoResponseMessageHandler (
						experiment, requester
						)
				);
	}
	
	protected Message createCanAddDataShareMessage (
			Node from, 
			Node to,
			CanAddDataShareRequester requester
			)
	{
		return createMessage(
				MessageType.CanAddDataShare, 
				from,
				to, 
				MessagePurpose.Caching,				
				new CanAddDataShareMessageHandler (
						experiment, requester
						)
				);
	}
	
	protected Message createCanAddDataShareResponseMessage (
			Node from, 
			Node to,
			boolean success,
			CanAddDataShareRequester requester
			)
	{
		return createMessage(
				((success)?(MessageType.Affermative):(MessageType.Negative)), 
				from,
				to, 
				MessagePurpose.Caching,				
				new CanAddDataShareResponseMessageHandler (
						experiment, requester
						)
				);
	}
	
	protected Message createDownloadAndCacheShareMessage (
			Node from, 
			Node to,
			Cluster hostCluster,
			DataObject dobj,
			int shareIndex,
			ShareCacheRequester requester
			)
	{
		return createMessage(
				MessageType.DownloadAndCacheShare, 
				from,
				to, 
				MessagePurpose.Caching,				
				new DownloadAndCacheShareMessageHandler (
						experiment, hostCluster, dobj, shareIndex, requester
						)
				);
	}
	
	protected Message createDownloadAndCacheShareResponseMessage (
			Node from, 
			Node to,
			boolean success,
			DataObject dobj,
			int shareIndex,
			ShareCacheRequester requester
			)
	{
		return createMessage(
				((success)?(MessageType.Affermative):(MessageType.Negative)), 
				from, to, MessagePurpose.Caching,				
				new DownloadAndCacheShareResponseMessageHandler (
						experiment, dobj, shareIndex, requester
						)
				);
	}
	
	protected Message createActivateCachedShareMessage (
			Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			ActivateCacheShareRequester requester
			)
	{
		return createMessage(
				MessageType.ActivateCachedShare, from, to, 
				MessagePurpose.Caching,				
				new ActivateCacheShareMessageHandler (
						experiment, dobj, shareIndex, requester
						)
				);
	}
	
	protected Message createShareActivatedResponseMessage (
			Node from, 
			Node to,
			boolean success,
			DataObject dobj,
			int shareIndex,
			ActivateCacheShareRequester requester
			)
	{
		return createMessage(
				((success)?(MessageType.Affermative):(MessageType.Negative)), 
				from, to, MessagePurpose.Caching,				
				new ShareActivatedResponseMessageHandler (
						experiment, dobj, shareIndex, requester
						)
				);
	}
	
	

	protected Message createRequestDataObjectMessage (
			Node from, 
			Node to,
			DataObject dobj,
			MessagePurpose purpose,
			DataObjectDownloader downloader
			)
	{
		return createMessage(
				MessageType.RequestDataObject, 
				from, to, purpose,				
				new RequestDataObjectMessageHandler (
						experiment, dobj, purpose, downloader
						)
				);
	}
	
	protected Message createRequestDataObjectLocalMessage (
			Node from, 
			Node to,
			DataObject dobj,
			MessagePurpose purpose,
			DataObjectDownloader downloader
			)
	{
		return createMessage(
				MessageType.RequestDataObjectLocal, 
				from, to, purpose,				
				new RequestDataObjectLocalMessageHandler (
						experiment, from, dobj, purpose, downloader
						)
				);
	}
	
	protected Message createDataObjectTransferMessage (
			Node from, 
			Node to,
			DataObject dobj,
			MessagePurpose purpose,
			DataObjectDownloader downloader
			)
	{
		return createMessage(
				MessageType.DataObjectTransfer, 
				from, to, dobj, purpose, 
				new RequestDataObjectResponseMessageHandler (
						experiment, downloader
						)
				);
	}
	
	protected Message createDoesNotHaveDataObjectMessage (
			Node from, 
			Node to,
			MessagePurpose purpose,
			DataObjectDownloader downloader
			)
	{
		return createMessage(
				MessageType.Negative, 
				from, to, purpose,				
				new RequestDataObjectResponseMessageHandler (
						experiment, downloader
						)
				);
	}
	
	protected Message createCacheDataObjectMessage (
			Node from, 
			Node to,
			Cluster shouldHaveIt,
			DataObject dobj,
			DataObjectCacheRequester requester
			)
	{
		return createMessage(
				MessageType.DownloadAndCacheDataObject, 
				from, to, MessagePurpose.Caching,				
				new DownloadAndCacheDataObjectMessageHandler (
						experiment, shouldHaveIt, dobj, requester
						)
				);
	}
	
	protected Message createDownloadAndCacheDataObjectResponseMessage (
			Node from, 
			Node to,
			boolean success,
			DataObject dobj,
			DataObjectCacheRequester requester
			)
	{
		return createMessage(
				((success)?(MessageType.Affermative):(MessageType.Negative)),  
				from, to, MessagePurpose.Caching,				
				new DownloadAndCacheDataObjectResponseMessageHandler (
						experiment, dobj, requester
						)
				);
	}*/
	
	
	
	protected Message createMessage (
			MessageType type, 
			Node from, 
			Node to, 
			MessagePurpose purpose, 
			MessageHandler handler
			)
	{
		long msgSize = experiment.modules().messageSizeCalculator().
			calculateSize(type);
		
		Pair<Double,Double> costs = getCommunicationCosts(
				from, to, msgSize
				);
		
		// OK, queue up message!
		Message m = new Message(
				experiment.getNewMessageId(),
				type, purpose, from, to, msgSize, costs.first, costs.second,
				handler
				);
		return m;
	}
	
	protected Message createMessage (
			MessageType type, 
			Node from,
			Node to, 
			int elementCount,
			int elementSize,
			MessagePurpose purpose, 
			MessageHandler handler
			)
	{
		long msgSize = experiment.modules().messageSizeCalculator().
			calculateSize(type, elementCount, elementSize);
		
		Pair<Double,Double> costs = getCommunicationCosts(
				from, to, msgSize
				);
		
		// OK, queue up message!
		Message m = new Message(
				experiment.getNewMessageId(),
				type, purpose, from, to, msgSize, costs.first, costs.second,
				handler
				);
		return m;
	}
	
	protected Message createMessage (
			MessageType type, 
			Node from,
			Node to,
			DataObject dobj,
			int elementCount,
			int elementSize,
			MessagePurpose purpose, 
			MessageHandler handler
			)
	{
		long msgSize = experiment.modules().messageSizeCalculator().
			calculateSize(type, dobj, elementCount, elementSize);
		
		Pair<Double,Double> costs = getCommunicationCosts(
				from, to, msgSize
				);
		
		// OK, queue up message!
		Message m = new Message(
				experiment.getNewMessageId(),
				type, purpose, from, to, msgSize, costs.first, costs.second,
				handler
				);
		return m;
	}
	
	protected Message createMessage (
			MessageType type, 
			Node from,
			Node to, DataObject dobj, 
			MessagePurpose purpose, 
			MessageHandler handler
			)
	{
		long msgSize = experiment.modules().messageSizeCalculator().
			calculateSize(type, dobj);
		
		Pair<Double,Double> costs = getCommunicationCosts(
				from, to, msgSize
				);
		
		// OK, queue up message!
		Message m = new Message(
				experiment.getNewMessageId(),
				type, purpose, from, to, msgSize, costs.first, costs.second,
				handler
				);
		return m;
	}
	
	protected Message createMessage (
			MessageType type, 
			Node from,
			Node to, 
			BloomFilter<DataObject> bf, 
			MessagePurpose purpose, 
			MessageHandler handler
			)
	{
		long msgSize = experiment.modules().messageSizeCalculator().
			calculateSize(type, bf);
		
		Pair<Double,Double> costs = getCommunicationCosts(
				from, to, msgSize
				);
		
		// OK, queue up message!
		Message m = new Message(
				experiment.getNewMessageId(),
				type, purpose, from, to, msgSize, costs.first, costs.second,
				handler
				);
		return m;
	}
	
	
	
	//Added
	protected Message createRequestVersionMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, 
			VersionReader versionReader)
	{
		return createMessage(MessageType.RequestVersion, from, to, purpose, 
				new RequestVersionMessageHandler(experiment, dobj, versionReader));
	}
	
	protected Message createVersionResponseMessage(Node from, Node to, TreeSet<Version> versions, 
			MessagePurpose purpose, VersionReader versionReader)
	{
		//System.out.println(versions.size());
		return createMessage(MessageType.VersionResponse, from, to, versions.size(),
				2, purpose,	new VersionResponseMessageHandler(experiment, versions, versionReader));
	}
	
	protected Message createShareUpdateMessage(Node from, Node to, DataObject dobj, Version ver, 
			Version stableVersion, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdate, from, to, dobj, purpose, 
				new ShareUpdateMessageHandler(experiment, dobj, ver, stableVersion, downloader));
	}
	
	protected Message createRequestLatestShareMessage(Node from, Node to, DataObject dobj, 
			MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.RequestLatestShare, from, to, purpose, 
				new RequestLatestShareMessageHandler(experiment, dobj, reader));
	}
	
	protected Message createShareDownloadMessage(Node from, Node to, DataObject dobj, Version version, 
			MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.ShareDownload, from, to, dobj, purpose, 
				new ShareDownloadMessageHandler(experiment, version, reader));
	}
	
	protected Message createRequestPreviousShareMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, VersionReader reader, int round)
	{
		return createMessage(MessageType.RequestPreviousShare, from, to, purpose, 
				new RequestPreviousShareMessageHandler(experiment, dobj, version, reader, round));
	}
	
	protected Message createShareUpdateV2Message(Node from, Node to, DataObject dobj, Version ver, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV2, from, to, dobj, purpose, 
				new ShareUpdateV2MessageHandler(experiment, dobj, ver, downloader));
	}
	
	protected Message createRequestLatestShareV2Message(Node from, Node to, DataObject dobj, 
			MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.RequestLatestShareV2, from, to, purpose, 
				new RequestLatestShareV2MessageHandler(experiment, dobj, reader));
	}
	
	protected Message createRequestPreviousShareV2Message(Node from, Node to, DataObject dobj,
			int n, MessagePurpose purpose, VersionReader reader, int round)
	{
		return createMessage(MessageType.RequestPreviousShareV2, from, to, purpose, 
				new RequestPreviousShareV2MessageHandler(experiment, dobj, n, reader, round));
	}
	
	protected Message createRequestVersionV3Message(Node from, Node to,	MessagePurpose purpose, 
			VersionReader versionReader)
	{
		return createMessage(MessageType.RequestVersionV3, from, to, purpose,
				new RequestVersionV3MessageHandler(experiment, versionReader));
	}
	
	protected Message createVersionResponseV3Message(Node from, Node to, int seq, 
			MessagePurpose purpose, VersionReader versionReader)
	{
		return createMessage(MessageType.VersionResponseV3, from, to, purpose, 
				new VersionResponseV3MessageHandler(experiment, seq, versionReader));
	}
	
	protected Message createShareUpdateV3Message(Node from, Node to, DataObject dobj, Version version, 
			MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV3, from, to, dobj, purpose, 
				new ShareUpdateV3MessageHandler(experiment, dobj, version, downloader));
	}
	
	protected Message createUpdateSuccessV3Message(Node from, Node to, DataObject dobj, Version version, 
			MessagePurpose purpose)
	{
		return createMessage(MessageType.UpdateSuccessV3, from, to, purpose, 
				new UpdateSuccessV3MessageHandler(experiment, dobj, version));
	}
	
	protected Message createRemoveOldShareV3Message(Node from, Node to, DataObject dobj, Version version, 
			MessagePurpose purpose)
	{
		return createMessage(MessageType.RemoveOldVersionV3, from, to, purpose, 
				new RemoveOldShareV3MessageHandler(experiment, dobj, version));
	}
	
	protected Message createShareUpdateV4Message(Node from, Node to, DataObject dobj, int shareIndex, 
			Version version, boolean tranx, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV4, from, to, dobj, purpose, 
				new ShareUpdateV4MessageHandler(experiment, dobj, shareIndex, version, tranx, downloader));
	}
	
	protected Message createShareUpdateV4ResponseMessage(Node from, Node to, DataObject dobj, int shareIndex,
			boolean tranx, boolean affirm, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV4Response, from, to, purpose,
				new ShareUpdateV4ResponseMessageHandler(experiment, dobj, shareIndex, tranx, affirm, downloader));
	}
	
	protected Message createRequestVersionQuorumMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, 
			VersionReader versionReader)
	{
		return createMessage(MessageType.RequestVersionQuorum, from, to, purpose, 
				new RequestVersionQuorumMessageHandler(experiment, dobj, versionReader));
	}
	
	protected Message createVersionResponseQuorumMessage(Node from, Node to, int seq, int index, 
			MessagePurpose purpose, VersionReader versionReader)
	{
		return createMessage(MessageType.VersionResponseQuorum, from, to, purpose, 
				new VersionResponseQuorumMessageHandler(experiment, seq, index, versionReader));
	}
	
	protected Message createShareUpdateV5Message(Node from, Node to, DataObject dobj, Version ver, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV5, from, to, dobj, purpose, 
				new ShareUpdateV5MessageHandler(experiment, dobj, ver, downloader));
	}
	
	protected Message createShareUpdateV5ResponseMessage(Node from, Node to, DataObject dobj, int index, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV5Response, from, to, purpose, 
				new ShareUpdateV5ResponseMessageHandler(experiment, dobj, index, downloader));
	}
	
	/*protected Message createReadRequestMessage(Node from, Node to, DataObject dobj, DataRequest dr, 
			MessagePurpose purpose, LinkedList<Node> rns, CoordinatorSelector selector, ShareDownloader downloader, LockProcessor processor){
		return createMessage(MessageType.ReadRequest, from, to, purpose,
				new ReadRequestMessageHandler(experiment, dobj, dr, rns, selector, downloader, processor));
	}
	
	protected Message createReadRequestResponseMessage(Node from, Node to, DataObject dobj, DataRequest dr, MessagePurpose purpose, LinkedList<Node> nodes, CoordinatorSelector selector, ShareDownloader downloader, LockProcessor processor){
		return createMessage(MessageType.ReadRequestResponse, from, to, purpose,
				new ReadRequestResponseMessageHandler(experiment, dobj, dr, nodes, selector, downloader, processor));
//	}*/
	
	/*protected Message createWriteRequestResponseMessage(Node from, Node to, DataObject dobj, DataRequest dr, MessagePurpose purpose, LinkedList<Node> nodes, CoordinatorSelector selector, ShareDownloader downloader){
		return createMessage(MessageType.WriteRequestResponse, from, to, purpose,
				new WriteRequestResponseMessageHandler(experiment, dobj, dr, nodes, selector, downloader));
	}*/
	
	protected Message createRequestLockMessage(Node from, Node to, DataObject dobj, DataRequest dr, MessagePurpose purpose, LockProcessor lp){
		return createMessage(MessageType.RequestLock, from, to, purpose,
				new RequestLockMessageHandler(experiment, dobj, dr, lp));
	}
	
	protected Message createRequestLockResponseMessage(Node from, Node to, DataRequest dr, boolean success, MessagePurpose purpose, LockProcessor lp){
		return createMessage(MessageType.RequestLockResponse, from, to, purpose,
				new RequestLockResponseMessageHandler(experiment, dr, success, lp));
	}
	
	protected Message createRequestShareV2Message(Node from, 
			Node to,
			DataObject dobj,
			int shareIndex,
			ShareDownloader shareDownloader
			)
	{
		return createMessage(
				MessageType.RequestShareV2, from,
				to, MessagePurpose.DataObjectRequest,
				new RequestShareV2MessageHandler (
						experiment,
						dobj,
						shareIndex,
						shareDownloader
						)
				);
	}
	
	protected Message createReleaseLockMessage(Node from, Node to, MessagePurpose purpose, DataRequest request, LockProcessor processor){
		return createMessage(MessageType.ReleaseLock, from, to, purpose,
				new ReleaseLockMessageHandler(experiment, request, processor));
	}
	
	protected Message createShareUpdateV6Message(Node from, Node to, DataObject dobj, Version ver, 
			MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV6, from, to, dobj, purpose, 
				new ShareUpdateV6MessageHandler(experiment, dobj, ver, downloader));
	}
	
	protected Message createShareUpdateV6ResponseMessage(Node from, Node to, DataObject dobj, int index, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateV6Response, from, to, purpose, 
				new ShareUpdateV6ResponseMessageHandler(experiment, dobj, index, downloader));
	}
	
	protected Message createRequestDesignatedShareMessage(Node from, Node to, DataObject dobj, Version version, MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.RequestDesignatedShare, from, to, purpose, 
				new RequestDesignatedShareMessageHandler(experiment, dobj, version, reader));
	}
	
	protected Message createRequestMaxTSMessage(Node from, Node to,	DataObject dobj, MessagePurpose purpose, 
			VersionReader versionReader)
	{
		return createMessage(MessageType.RequestMaxTS, from, to, purpose,
				new RequestMaxTSMessageHandler(experiment, dobj, versionReader));
	}
	
	protected Message createRequestLatestAndVersionMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, 
			VersionReader versionReader)
	{
		return createMessage(MessageType.RequestLatestAndVersion, from, to, purpose, 
				new RequestLatestAndVersionMessageHandler(experiment, dobj, versionReader));
	}
	
	protected Message createLatestAndVersionResponseMessage(Node from, Node to, DataObject dobj, TreeSet<Version> versions, 
			Version version, MessagePurpose purpose, VersionReader versionReader){
		return createMessage(MessageType.LatestAndVersionResponse, from, to, dobj, versions.size(),
				2, purpose,	new LatestAndVersionResponseMessageHandler(experiment, versions, version, versionReader));
	}
	
	protected Message createRequestLatestAndVersionV2Message(Node from, Node to, DataObject dobj, MessagePurpose purpose, 
			VersionReader versionReader)
	{
		return createMessage(MessageType.RequestLatestAndVersionV2, from, to, purpose, 
				new RequestLatestAndVersionV2MessageHandler(experiment, dobj, versionReader));
	}
	
	protected Message createLatestAndVersionResponseV2Message(Node from, Node to, DataObject dobj, LinkedList<Version> versions, 
			Version version, MessagePurpose purpose, VersionReader versionReader){
		return createMessage(MessageType.LatestAndVersionResponseV2, from, to, dobj, versions.size(),
				2, purpose,	new LatestAndVersionResponseV2MessageHandler(experiment, versions, version, versionReader));
	}
	
	//rep
	protected Message createRequestUpdateLockMessage(Node from, Node to, DataObject dobj, DataRequest dr, MessagePurpose purpose, LockProcessor lp){
		return createMessage(MessageType.RequestUpdateLock, from, to, purpose,
				new RequestUpdateLockMessageHandler(experiment, dobj, dr, lp));
	}
	protected Message createUpdateLockResponseMessage(Node from, Node to, MessagePurpose purpose, LockProcessor lp){
		return createMessage(MessageType.UpdateLockResponse, from, to, purpose,
				new UpdateLockResponseMessageHandler(experiment, lp));
	}
	protected Message createReleaseUpdateLockMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose){
		return createMessage(MessageType.ReleaseUpdateLock, from, to, purpose,
				new ReleaseUpdateLockMessageHandler(experiment, dobj));
	}
	protected Message createRequestVersionRepMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, Boolean ru, VersionReader reader){
		return createMessage(MessageType.RequestVersionRep, from, to, purpose,
				new RequestVersionRepMessageHandler(experiment, dobj, ru, reader));
	}
	protected Message createTurnVersionRepOnMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, int version, Boolean onoff, VersionReader reader){
		return createMessage(MessageType.TurnVersionRepOn, from, to, purpose,
				new TurnVersionRepOnMessageHandler(experiment, dobj, version, onoff, reader));
	}
	protected Message createRequestDesignatedShareRepMessage(Node from, Node to, DataObject dobj, int version, MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.RequestDesignatedShareRep, from, to, purpose, 
				new RequestDesignatedShareRepMessageHandler(experiment, dobj, version, reader));
	}
	
	//svr;
	protected Message createReadCompleteMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, VersionReader reader, boolean multi)
	{
		return createMessage(MessageType.ReadComplete, from, to, purpose, 
				new ReadCompleteMessageHandler(experiment, dobj, reader, multi));
	}
	
	//two level ordered version
	protected Message createRequestVersionTLOMessage(Node from, Node to, DataObject dobj, MessagePurpose purpose, Boolean ru, LinkedList<Node> rs, VersionReader reader){
		return createMessage(MessageType.RequestVersionTLO, from, to, purpose,
				new RequestVersionTLOMessageHandler(experiment, dobj, ru, rs, reader));
	}
	protected Message createShareUpdateTLOMessage(Node from, Node to, DataObject dobj, 
			Version version, boolean tranx, boolean repair, LinkedList<Cluster> ring, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateTLO, from, to, dobj, purpose, 
				new ShareUpdateTLOMessageHandler(experiment, dobj, version, tranx, repair, ring, downloader));
	}
	protected Message createUpdateSuccessTLOMessage(Node from, Node to, DataObject dobj, Version version, 
			MessagePurpose purpose)
	{
		return createMessage(MessageType.UpdateSuccessTLO, from, to, purpose, 
				new UpdateSuccessTLOMessageHandler(experiment, dobj, version));
	}
	protected Message createSVRMessage(Node from, Node to, DataObject dobj, Version version, MessagePurpose purpose)
	{
		return createMessage(MessageType.SVR, from, to, purpose, 
				new SVRMessageHandler(experiment, dobj, version));
	}
	protected Message createRequestDesignatedShareTLOMessage(Node from, Node to, DataObject dobj, Version version, Node requester, MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.RequestDesignatedShareTLO, from, to, purpose, 
				new RequestDesignatedShareTLOMessageHandler(experiment, dobj, version, requester, reader));
	}
	protected Message createReadCompleteTLOMessage(Node from, Node to, DataObject dobj, boolean global, MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.ReadCompleteTLO, from, to, purpose, 
				new ReadCompleteTLOMessageHandler(experiment, dobj, global, reader));
	}
	
	//cvs
	protected Message createShareUpdateCVSMessage(Node from, Node to, DataObject dobj, 
			Version version, MessagePurpose purpose, ShareDownloader downloader)
	{
		return createMessage(MessageType.ShareUpdateCVS, from, to, dobj, purpose, 
				new ShareUpdateCVSMessageHandler(experiment, dobj, version, downloader));
	}
	
	//osv
	protected Message createShareUpdateOSVMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, ShareDownloader downloader){
		return createMessage(MessageType.ShareUpdateOSV, from, to, dobj, purpose,
				new ShareUpdateOSVMessageHandler(experiment, dobj, version, downloader));
	}
	protected Message createSendMACMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, ShareDownloader downloader){
		return createMessage(MessageType.SendMAC, from, to, purpose,
				new SendMACMessageHandler(experiment, dobj, version, downloader));
	}
	protected Message createRequestLCShareMessage(Node from, Node to, DataObject dobj,
			MessagePurpose purpose, VersionReader reader){
		return createMessage(MessageType.RequestLCShare, from, to, purpose,
				new RequestLCShareMessageHandler(experiment, dobj, reader));
	}
	protected Message createRequestShareWithVersionMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, VersionReader reader, boolean ir){
		return createMessage(MessageType.RequestShareWithVersion, from, to, purpose,
				new RequestShareWithVersionMessageHandler(experiment, dobj, version, reader, ir));
	}
	
	//hen
	protected Message createShareUpdateHenMessage(Node from, Node to, DataObject dobj, 
			MessagePurpose purpose, VersionReader reader)
	{
		return createMessage(MessageType.ShareUpdateHen, from, to, dobj, purpose,
				new ShareUpdateHenMessageHandler(experiment, dobj, reader));
	}
	protected Message createMACResponseMessage(Node from, Node to, DataObject dobj, Integer ts, 
			MessagePurpose purpose, VersionReader reader){
		return createMessage(MessageType.MACResponse, from, to, dobj, purpose,
				new MACResponseMessageHandler(experiment, ts, reader));
	}
	protected Message createSendVersionMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, VersionReader reader){
		return createMessage(MessageType.SendVersion, from, to, purpose,
				new SendVersionMessageHandler(experiment, dobj, version, reader));
	}
	protected Message createSendMACHenMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, ShareDownloader downloader){
		return createMessage(MessageType.SendMACHen, from, to, dobj, purpose,
				new SendMACHenMessageHandler(experiment, dobj, version, downloader));
	}
	protected Message createRequestLCVersionMessage(Node from, Node to, DataObject dobj,
			MessagePurpose purpose, VersionReader reader){
		return createMessage(MessageType.RequestLCVersion, from, to, purpose,
				new RequestLCVersionMessageHandler(experiment, dobj, reader));
	}
	protected Message createLCVersionResponseMessage(Node from, Node to, Version version, 
			MessagePurpose purpose, VersionReader reader){
		return createMessage(MessageType.LCVersionResponse, from, to, purpose,
				new LCVersionResponseMessageHandler(experiment, version, reader));
	}
		
	//avid
	protected Message createShareUpdateAVIDMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, ShareDownloader downloader){
		return createMessage(MessageType.ShareUpdateAVID, from, to, dobj, purpose,
				new ShareUpdateAVIDMessageHandler(experiment, dobj, version, downloader));
	}
	protected Message createSendMACAVIDMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, ShareDownloader downloader){
		return createMessage(MessageType.SendMACAVID, from, to, dobj, purpose,
				new SendMACAVIDMessageHandler(experiment, dobj, version, downloader));
	}
	
	//svr
	protected Message createRequestVersionNumberMessage(Node from, Node to, DataObject dobj,
			Version version, int round, MessagePurpose purpose, SVRHandler handler){
		return createMessage(MessageType.RequestVersionNumber, from, to, purpose,
				new RequestVersionNumberMessageHandler(experiment, dobj, version, handler, round));
	}
	protected Message createVersionNumberResponseMessage(Node from, Node to, Version version, int round, 
		 MessagePurpose purpose, SVRHandler handler){
		return createMessage(MessageType.VersionNumberResponse, from, to, purpose,
				new VersionNumberResponseMessageHandler(experiment, version, handler, round));
	}
	protected Message createBoundaryVersionMessage(Node from, Node to, DataObject dobj,
			Version version, MessagePurpose purpose, SVRHandler handler){
		return createMessage(MessageType.BoundaryVersion, from, to, purpose,
				new BoundaryVersionMessageHandler(experiment, dobj, version, handler));
	}
	protected Message createRequestVersionShareMessage(Node from, Node to, DataObject dobj,
			Version version, int round, MessagePurpose purpose, SVRHandler handler){
		return createMessage(MessageType.RequestVersionShare, from, to, purpose,
				new RequestVersionShareMessageHandler(experiment, dobj, version, handler, round));
	}
	protected Message createVersionShareResponseMessage(Node from, Node to, DataObject dobj, Version version, int round, 
		 MessagePurpose purpose, SVRHandler handler){
		return createMessage(MessageType.VersionShareResponse, from, to, dobj, purpose,
				new VersionShareResponseMessageHandler(experiment, version, handler, round));
	}
}